package it.wyx.kafkaclients.producer;

import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.common.serialization.StringSerializer;

import java.util.Properties;

/**
 * 生产者异步发送-demo
 * @author wyx
 * @version V 1
 * @Date 2022/10/30
 */
public class ProducerAsync {
	private static final String BROKER_LIST = "localhost:9092";
	private static final String TOPIC = "test-topic";

	/**
	 * 创建 kafka 生产者的配置对象
	 */
	public static Properties initProducerConfig() {
		Properties properties = new Properties();
		// 连接地址
		properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, BROKER_LIST);
		// 键、值的序列化方式
		properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
		properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
		properties.put(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG, true);
		return properties;
	}

	public static void main(String[] args) {
		// 创建 kafka 生产者对象
		KafkaProducer<String, String> producer = new KafkaProducer<>(initProducerConfig());

		// 调用 send 方法,发送消息
		long l1 = System.currentTimeMillis();
		for (int i = 0; i < 5000; i++) {
			producer.send(new ProducerRecord<>(TOPIC, "hello kafka " + i));
		}
		long l2 = System.currentTimeMillis();
		System.out.println("ProducerAsync：l2 - l1 = " + (l2 - l1));
		// 5、关闭资源
		producer.close();
	}
}